package com.bw.utils;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDC {
    public static void main(String[] args) throws Exception {

        System.setProperty("HADOOP_USER_NAME","root");
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK，如果想要做到断点续传，需要从Checkpoint或者Savepoint启动程序
        //2.1 开启CK 并指定状态后端为FS   memory  fs  rocksdb
        env.setStateBackend(new FsStateBackend("hdfs://hadoop100:9000/gmall-flink-2021/ck"));
        //2.2 开启 Checkpoint,每隔 5 秒钟做一次 CK  头和头
        env.enableCheckpointing(5000);
        //2.3 指定 CK 的一致性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);//尾和头
        //2.4 指定从 CK 自动重启策略 ,老版本需要设置,因为1.10版本重启次数是int最大值
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000L));

        //3.通过FlinkCDC 构建SourceFunction并读取数据
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop100")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("aa")//可以同时读多个库
                //可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式
                .tableList("aa.base_attr_value")
                .deserializer(new StringDebeziumDeserializationSchema())
                //initial 初始化，
                .startupOptions(StartupOptions.initial())
                //.startupOptions(StartupOptions.latest())  //latest 是直接读最新的数据，而不会去打印老的数据
                .build();

        DataStreamSource<String> streamSource = env.addSource(sourceFunction);

        //3.打印数据
        streamSource.print();

        //4.启动任务
        env.execute();

    }
}
